package io.jjz;

import io.jjz.utils.ByteUtil;
import com.google.common.collect.Sets;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.NavigableMap;
import java.util.Set;

/**
 * Author       : 蒋俊钊 80256973
 * Date         : 2019/2/17
 * Copyright    (C) ChinaMerchantsBank
 */
public class IndexCoprocessor extends BaseRegionObserver {

    // 要排序的字段
    private static final Set<String> orderFieldSet = Sets.newHashSet("YTA", "YTN", "TMN", "DTA", "DAP", "DTN", "DNP", "MTA", "MTN", "M2AP", "M2NP", "MAP", "MNP");

    // 数据列族
    private static final byte[] DATA_COL = Bytes.toBytes("d");
    // 索引列族
    private static final byte[] INDEX_COL = Bytes.toBytes("i");

    private RegionCoprocessorEnvironment env = null;

    @Override
    public void start(CoprocessorEnvironment e) throws IOException {
        env = (RegionCoprocessorEnvironment) e;
    }

    /**
     * 每插入一条记录后，再插入对该记录的二级索引
     */
    @Override
    public void postPut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability) throws IOException {
        TableName tableName = env.getRegionInfo().getTable();
        Table table = env.getTable(tableName);

        NavigableMap<byte[], List<Cell>> familyCellMap = put.getFamilyCellMap();
        // 获取主数据列族d
        List<Cell> cells = familyCellMap.get(DATA_COL);
        if (cells == null || cells.size() == 0) {
            // 当不存在对于d列的操作的时候则不做任何的处理，直接放行即可
            //LOG.info("Column A don't have operation, just do it.");
            return;
        }

        // 遍历数据列族的各个列
        for (Cell c : cells) {
            byte[] qualifier = CellUtil.cloneQualifier(c);
            if (orderFieldSet.contains(new String(qualifier))) {
                // 构建二级索引  RegionStartKey-索引名(列名)-索引键(列值)-索引值(rowkey)
                //byte[] regionStartKey = env.getRegionInfo().getStartKey();

                // 二级索引: 索引名(列名)-日期-分行号-类型-索引键(列值)-索引值(商户号)
                // 该条记录的rowkey字符串
                String rowkey = new String(CellUtil.cloneRow(c));
                String[] split = rowkey.split("\\|");

                // 主数据的rowkey要符合能拆分成4段，才为其构建索引
                if (split.length == 4) {
                    // 列值
                    byte[] value = CellUtil.cloneValue(c);
                    // value要是int或double类型，其他为字符串
                    byte[] indexRowKey = ByteUtil.mergerBytes(qualifier, split[0].getBytes(), split[1].getBytes(), split[2].getBytes(), value, split[3].getBytes());
                    Put indexPut = new Put(indexRowKey);
                    indexPut.addColumn(INDEX_COL, null, "/".getBytes());
                    table.put(indexPut);
                }
            }
        }
        table.close();
    }

    /**
     * todo:索引表禁用自动split
     * 数据表split时要修改索引表的rowkey
     */
    @Override
    public void preSplit(ObserverContext<RegionCoprocessorEnvironment> c, byte[] splitRow) throws IOException {
        super.preSplit(c, splitRow);
    }
}
